#pragma once

#include <vector>
#include <cstdint>
#include <string>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <signal.h>
#include <fcntl.h>
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <array>
#include <thread>
#include <mutex>
#include <memory>
#include <any>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <condition_variable>

#include "Log.hpp"

class Channel;
class Poller;
class EventLoop;
class Timer;

class Buffer
{
    static const size_t DEFAULT_BUFFER_INITIAL_SIZE = 2048;
    static const size_t npos = -1;
private:
    std::vector<uint8_t> __buffer; // 缓冲区
    size_t __buffer_start_offset; // 缓冲区读开始位置
    size_t __buffer_end_offset; // 缓冲区写开始位置
    size_t __buffer_size; // 缓冲区元素大小

private:

    // 判断缓冲区是否已满
    bool is_full() { return __buffer.size() == __buffer_size; }

    // 重新buffer设置大小
    void resize(size_t n);

    // 扩容，确保有大于等于n的空间
    void extend_enough_space(size_t n);

public:

    Buffer(size_t n = DEFAULT_BUFFER_INITIAL_SIZE) : __buffer(n) {}
    Buffer(const Buffer& buffer)
    {
        __buffer = buffer.__buffer;
        __buffer_start_offset = buffer.__buffer_start_offset;
        __buffer_end_offset = buffer.__buffer_end_offset;
        __buffer_size = __buffer_size;
    }

    // 写数据
    void write(const void* data, size_t data_size);
    void write(const std::string& data) { write(data.c_str(), data.size()); }
    void write(const std::vector<uint8_t>& data) { write(data.data(), data.size()); }
    void write(const Buffer& data);


    // 读数据
    std::vector<uint8_t> read(size_t data_size = npos);
    std::string read_to_string(size_t data_size = npos)
    {
        auto data = read(data_size);
        return std::string(data.begin(), data.end());
    }
    std::string get_line();

    // 获取可读空间大小
    const size_t readable_size() { return __buffer_size; }

    // 清空缓冲区
    void clear();
};

void Buffer::resize(size_t n)
{
    std::vector<uint8_t> new_buffer(n);
    for (int i = 0; __buffer_start_offset != __buffer_end_offset; ++i)
    {
        if (__buffer_start_offset >= __buffer.size())
            __buffer_start_offset = 0;
        new_buffer[i] = __buffer[__buffer_start_offset++];
    }
    __buffer.swap(new_buffer);
}

void Buffer::extend_enough_space(size_t n)
{
    size_t new_space = __buffer.size();

    while (new_space < n)
    {
        new_space <<= 1;
    }
    resize(new_space);
}

// 写数据
void Buffer::write(const void* data, size_t data_size)
{
    if (data_size == 0) return;
    // 1. 判断缓冲区是否已满
    if (is_full() || data_size + __buffer_size > __buffer.size())
    {
        extend_enough_space(data_size + __buffer_size);
    }

    for (int i = 0; i < data_size; ++i)
    {
        if (__buffer_end_offset >= __buffer.size())
            __buffer_end_offset = 0;
        __buffer[__buffer_end_offset++] = *((uint8_t*)data + i);
        ++__buffer_size;
    }
}

// 读数据
std::vector<uint8_t> Buffer::read(size_t data_size)
{
    std::vector<uint8_t> data;
    auto tmp_size = __buffer_size;
    for (int i = 0; i < data_size && i < tmp_size; ++i)
    {
        if (__buffer_start_offset >= __buffer.size())
            __buffer_start_offset = 0;
        data.push_back(__buffer[__buffer_start_offset++]);
        --__buffer_size;
    }
    return data;
}

// 清空缓冲区
void Buffer::clear()
{
    __buffer_start_offset = __buffer_end_offset = __buffer_size = 0;
}

void Buffer::write(const Buffer& data)
{
    if (data.__buffer_size + __buffer_size >= __buffer.size())
    {
        extend_enough_space(data.__buffer_size + __buffer_size);
    }
    for (int i = 0, data_offset = data.__buffer_start_offset; i < data.__buffer_size; ++i)
    {
        if (__buffer_end_offset >= __buffer.size())
            __buffer_end_offset = 0;
        if (data_offset >= data.__buffer.size())
            data_offset = 0;
        __buffer[__buffer_end_offset++] = data.__buffer[data_offset++];
        ++__buffer_size;
    }
}

std::string Buffer::get_line()
{
    std::string str;
    auto tmp_size = __buffer_size;
    for (int i = 0; i < tmp_size; ++i)
    {
        if (__buffer_start_offset >= __buffer.size())
            __buffer_start_offset = 0;
        if (__buffer[__buffer_start_offset] == '\n')
        {
            break;
        }
        str += __buffer[__buffer_start_offset++];
        --__buffer_size;
    }
    str += __buffer[__buffer_start_offset++];
    --__buffer_size;

    return str;
}



class Socket
{
    const static uint32_t MAX_LISTEN = 1024;
    const static int ERROR_SOCKET = -1;

private:
    int _sockfd;
public:
    Socket(): _sockfd(ERROR_SOCKET) {}
    Socket(int file_descriptor): _sockfd(file_descriptor) {}
    ~Socket() { Close(); }

    // 创建套接字
    bool Create();

    int getFd() { return _sockfd; }

    // 关闭套接字
    void Close()
    {
        if (_sockfd != ERROR_SOCKET)
            close(_sockfd);
        _sockfd = ERROR_SOCKET;
    }

    // 绑定套接字
    bool Bind(const std::string& ip, uint16_t port);

    // 监听套接字
    bool Listen(uint32_t backlog = MAX_LISTEN);

    // 向服务端发起连接
    bool Connect(const std::string& ip, uint16_t port);

    // 获取新连接
    int Accept();

    // 接收数据
    ssize_t Recv(void *buf, size_t len, int flags = 0);
    ssize_t RecvNonBlock(void *buf, size_t len) { return Recv(buf, len, MSG_DONTWAIT); }

    // 发送数据
    ssize_t Send(const void *buf, size_t len, int flags = 0);
    ssize_t SendNonBlock(void *buf, size_t len) { return Send(buf, len, MSG_DONTWAIT); }

    // 创建一个服务端监听
    bool CreateServer(uint16_t port, const std::string& ip = "0.0.0.0", bool setNoBlock = false);

    // 创建一个客户端连接
    bool CreateClient(uint16_t port, const std::string& ip);

    // 开启地址重用
    void ReuseAddress();

    // 设置为非阻塞
    void NonBlock();
};



class Channel
{
    using EventHandleType = std::function<void()>;
private:
    int _fd;
    uint32_t _monitor_events;       // 当前需要监控的事件
    uint32_t _trigger_events;       // 当前触发的事件
    // Poller* _poller;
    EventLoop *_loop;
    EventHandleType _read_handle;   // 可读 事件被触发的回调函数
    EventHandleType _write_handle;  // 可写 事件被触发的回调函数
    EventHandleType _error_handle;  // 错误 事件被触发的回调函数
    EventHandleType _close_handle;  // 关闭 事件被触发的回调函数
    EventHandleType _any_handle;    // 任意 事件的回调函数
public:

    Channel(int file_descriptor, EventLoop* event_loop)
        : _fd(file_descriptor)
        , _monitor_events(0)
        , _trigger_events(0)
        , _loop(event_loop) 
    {
        LOG(DEBUG, "Channel init");
    }

    void SetTriggerEvents(uint32_t events) { _trigger_events = events; }
    int getFd() { return _fd; }
    uint32_t getMonitorEvents() { return _monitor_events; }

    // 判断是否可读
    bool isReadable() { return _monitor_events & EPOLLIN; }
    // 判断是否可写
    bool isWriteable() { return _monitor_events & EPOLLOUT; }

    // 设置读事件回调
    void SetReadHandle(const EventHandleType& handle) { _read_handle = handle; }
    // 设置写事件回调
    void SetWriteHandle(const EventHandleType& handle) { _write_handle = handle; }
    // 设置错误事件回调
    void SetErrorHandle(const EventHandleType& handle) { _error_handle = handle; }
    // 设置关闭事件回调
    void SetCloseHandle(const EventHandleType& handle) { _close_handle = handle; }
    // 设置任意事件回调
    void SetEventHandle(const EventHandleType& handle) { _any_handle = handle; }


    // 设置事件监控

    // 启动读事件监控
    void StartReadEventMonitor();
    // 启动写事件监控
    void StartWriteEventMonitor();
    // 关闭读事件监控
    void CloseReadEventMonitor();
    // 关闭写事件监控
    void CloseWriteEventMonitor();
    // 关闭所有事件监控
    void CloseEventMonitor();

    // 移除事件监控
    void RemoveEvent();

    // 更新事件
    void UpdateEvent();

    // 触发事件调用
    void TriggerEvent();

};

class Poller
{
    static const size_t MAX_EPOLL_EVENTS = 1024;
private:
    int _epoll_fd;
    std::array<struct epoll_event, MAX_EPOLL_EVENTS> _events;
    std::unordered_map<int, Channel*> _channels;

private:
    /*
        op: 
        EPOLL_CTL_ADD 添加
        EPOLL_CTL_MOD 修改
        EPOLL_CTL_DEL 删除
    */
    void PollerCtl(Channel* channel, int op) noexcept;

    // 判断是否添对channnel的监控
    bool haveChannel(Channel* channel) noexcept;

public:
    Poller()
    {
        _epoll_fd = epoll_create(MAX_EPOLL_EVENTS);
        if (_epoll_fd < 0)
        {
            LOG(ERROR, "epoll create error");
        }
    }

    // 更新Event
    void UpdateEvent(Channel* channel) noexcept;

    // 移除 Event
    void RemoveEvent(Channel* channel) noexcept;

    // 开始监控，返回活跃连接
    void Poll(std::vector<Channel*>& active);
};

struct TimerTaskType
{
    using time_t = uint32_t;
    using taskid_t = uint64_t;
    using TaskFunc = std::function<void()>;
    using ReleaseFunc = std::function<void()>;
};

class TimerTask : public TimerTaskType
{
protected:
    taskid_t _id;       // 任务对象ID
    time_t _time_out; // 超时时间
    bool _is_canceled; // 任务被取消
    TaskFunc _call_back;
    ReleaseFunc _release; 
public:
    TimerTask(taskid_t id, time_t delay_time, const TaskFunc& function)
        : _id(id), _time_out(delay_time), _is_canceled(false), _call_back(function)
    {}
    ~TimerTask() 
    { 
        if (!_is_canceled) 
        {
            _call_back();
        }
        _release();
    }

    // 设置释放后执行的函数
    void setRelease(const ReleaseFunc& call_back) { _release = call_back; }

    //获取延时时间
    time_t getDelayTime() { return _time_out; }

    // 取消定时任务
    void canceled() { _is_canceled = true; }
};


class TimerWheel : public TimerTaskType
{
    using TaskPtr = std::shared_ptr<TimerTask>;
    using WeakTaskPtr = std::weak_ptr<TimerTask>;
public:
    static const int DEFAULT_DELAY_TIME = 60; // 默认延迟时间的最大刻度

protected:
    time_t _tick;
    time_t _maxest_delay_time;
    std::vector<std::vector<TaskPtr>> _time_wheel;
    std::unordered_map<taskid_t, WeakTaskPtr> _timers;
protected:

    void RemoveTimer(taskid_t id);

public:
    TimerWheel(time_t maxest_delay_time = DEFAULT_DELAY_TIME) 
        : _maxest_delay_time(maxest_delay_time)
        , _tick(0)
        , _time_wheel(DEFAULT_DELAY_TIME)
    {}

    // 增加定时任务
    void add(taskid_t id, time_t delay_time, const TaskFunc& call_back);

    // 刷新定时任务
    void flash(taskid_t id);

    // tick计时+1, 并执行指针指向的任务
    void runTick();

    // 取消定时任务
    void canceledTimer(taskid_t id);

    // 判断是否有该定时任务
    // 存在线程安全，必须在EventLoop内执行
    bool haveTimer(taskid_t id);
};

class Timer : public TimerTaskType
{
private:
    int _timerfd;
    EventLoop* _loop;
    std::unique_ptr<Channel> _channel;
    TimerWheel _timer_wheel;
private:

    // 创建定时器fd
    static int CraeteTimerFd();

    // 获取定时器
    void ReadTimerFd();

public:
    Timer(EventLoop* loop, TimerTaskType::time_t max_delay_time = TimerWheel::DEFAULT_DELAY_TIME)
        : _timerfd(CraeteTimerFd())
        , _loop(loop)
        , _channel(new Channel(_timerfd, _loop))
        , _timer_wheel(max_delay_time)
    {
        _channel->SetReadHandle([&]() -> void {
            ReadTimerFd();
            _timer_wheel.runTick();
        });
        _channel->StartReadEventMonitor();
    }

    // 增加定时任务
    void AddTimerTask(taskid_t id, time_t delay_time, const TaskFunc& handle);

    // 刷新定时任务
    void FlashTimerTask(taskid_t id);

    // 取消定时任务
    void CancelTimerTask(taskid_t id);

    bool HaveTimerTask(taskid_t id);
};

class EventLoop
{
    using func_t = std::function<void()>;
private:
    int _event_fd;
    std::thread::id _thread_id;
    std::unique_ptr<Channel> _event_channel;
    std::mutex _mutex;
    Poller  _poller;
    std::vector<func_t> _task_queue;
    Timer _timer;
public:
    EventLoop(TimerTaskType::time_t max_delay_time = TimerWheel::DEFAULT_DELAY_TIME);

    // 执行任务，若任务处于EventLoop线程中则执行，否则压入任务队列
    void RunInEvevtLoop(const func_t& task);

    // 将任务压入任务队列
    void PushIntoTaskQueue(const func_t& task);
    void UpdateEvent(Channel* channel) { return _poller.UpdateEvent(channel); }
    void RemoveEvent(Channel* channel) { return _poller.RemoveEvent(channel); }

    // 判断当前线程是否是EventLoop对应的线程
    bool IsEventLoopThread() { return _thread_id == std::this_thread::get_id(); }
    void AssertInThread() { assert(_thread_id == std::this_thread::get_id()); }

    // 开始运行EventLoop
    void StartLoop();

    // 增加定时任务
    void AddTimerTask(Timer::taskid_t id, Timer::time_t delay_time, const Timer::TaskFunc& handle) {
        _timer.AddTimerTask(id, delay_time, handle);
    }

    // 刷新定时任务
    void FlashTimerTask(Timer::taskid_t id) {
        _timer.FlashTimerTask(id);
    }

    // 取消定时任务
    void CancelTimerTask(Timer::taskid_t id) { 
        _timer.CancelTimerTask(id);
    }

    bool HaveTimerTask(Timer::taskid_t id) {
        return _timer.HaveTimerTask(id);
    }
    

private:
    void RunAllTask();
    static int CreateEventFd();
    void ReadeEventFd();
    void WeakUpEventFd();
};


class Connection : public std::enable_shared_from_this<Connection>
{

    enum Status{
        DISCONNECT,     // 连接关闭
        CONNECTING,     // 连接中, 待处理连接
        CONNECTED,      // 连接完成，完成连接设置
        DISCONNECTING   // 连接正在关闭
    };

public:
    using ConnectionPtr = std::shared_ptr<Connection>;
    using ConnectHandle = std::function<void(ConnectionPtr)>;
    using MessageHandle = std::function<void(ConnectionPtr, Buffer&)>;
    using CloseHandle = std::function<void(ConnectionPtr)>;
    using EventHandle = std::function<void(ConnectionPtr)>;
    
private:
    bool _is_inactive_connection;
    int _sockfd;
    uint64_t _connection_id;
    
    std::any _context; // 请求的接收处理（上下文）
    Status _status;
    EventLoop* _loop;
    std::unique_ptr<Channel> _channel;
    std::unique_ptr<Socket>  _socket;
    Buffer _in_buffer;
    Buffer _out_buffer;

    ConnectHandle _connected_handle;
    MessageHandle _message_handle;
    CloseHandle _close_handle;
    EventHandle _event_handle;
    CloseHandle _server_close_handle;

private:

    void Release()
    {
        _loop->RunInEvevtLoop([&]() -> void {

            LOG(DEBUG, "Release");

            _status = DISCONNECT;
            _channel->RemoveEvent();
            _socket->Close();
            if (_loop->HaveTimerTask(_connection_id))
                CloseInactiveRelease();
            
            if (_close_handle)
                _close_handle(shared_from_this());

            if (_server_close_handle)
                _server_close_handle(shared_from_this());
        });
    }

    void HandleRead()
    {
        uint8_t buf[65536] = { 0 };
        ssize_t ret = _socket->RecvNonBlock(buf, sizeof(buf));
        if (ret < 0)
        {
            return CloseConnection();
        }
        _in_buffer.write(buf, ret);
        if (_in_buffer.readable_size() > 0)
        {
            return _message_handle(shared_from_this(), _in_buffer);
        }

    }
    void HandleWrite()
    {
        auto send_date = _out_buffer.read();
        auto ret = _socket->SendNonBlock(send_date.data(), send_date.size());
        if (ret < 0)
        {
            if (_in_buffer.readable_size() > 0)
            {
                _message_handle(shared_from_this(), _in_buffer);
            }
            return Release();
        }

        if (_out_buffer.readable_size() == 0)
        {
            _channel->CloseWriteEventMonitor();
            if (_status == DISCONNECTING)
            {
                return Release();
            }
        }
    }
    void HandleClose()
    {
        if (_in_buffer.readable_size() > 0)
        {
            _message_handle(shared_from_this(), _in_buffer);
        }
        Release();
    }
    void HandleError()
    {
        HandleClose();
    }
    void HandleEvent()
    {
        if (_is_inactive_connection) 
            _loop->FlashTimerTask(_connection_id);
        if (_event_handle) 
            _event_handle(shared_from_this());
    }



public:
    Connection(EventLoop* loop, uint64_t connect_id, int sockfd)
        : _connection_id(connect_id)
        , _sockfd(sockfd)
        , _is_inactive_connection(false)
        , _loop(loop)
        , _status(CONNECTING)
        , _channel(new Channel(sockfd, loop))
        , _socket(new Socket(_sockfd))
    {
        _channel->SetReadHandle([&]() -> void { HandleRead(); });
        _channel->SetWriteHandle([&]() -> void { HandleWrite(); });
        _channel->SetEventHandle([&]() -> void { HandleEvent(); });
        _channel->SetCloseHandle([&]() -> void { HandleClose(); });
        _channel->SetErrorHandle([&]() -> void { HandleError(); });
    }

    int getFd() { return _sockfd; }
    uint64_t getId() { return _connection_id; }
    std::any getContext() { return _context; }

    void SetConnectHandle(ConnectHandle connected_handle) { _connected_handle = connected_handle; }
    void SetMessageHandle(MessageHandle message_handle) { _message_handle = message_handle; }
    void SetCloseHandle(CloseHandle close_handle) { _close_handle = close_handle; }
    void SetServerCloseHandle(CloseHandle server_close_handle) { _server_close_handle = server_close_handle; }
    void SetEventHandle(EventHandle event_handle) { _event_handle = event_handle; }
    void SetContext(std::any context) { _context.reset(); _context = context; }

    // 是否处于连接状态
    bool IsConnected() { return _status == CONNECTED; }

    // 连接建立完成就绪
    // 连接获取之后, 所处的状态下要进行的各种设置
    // 比如: 给channel设置事件回调, 启动读监控 .
    void Established()
    {
        _loop->RunInEvevtLoop([&]() -> void {
            assert(_status == CONNECTING);
            _status = CONNECTED;
            _channel->StartReadEventMonitor();
            if (_connected_handle) _connected_handle(shared_from_this());
        });
    }

    // 发送
    void Send(std::vector<uint8_t>& data)
    {
        _loop->RunInEvevtLoop([this, data]() -> void {
            if (_status == DISCONNECT) return;
            _out_buffer.write(data);
            if (!_channel->isWriteable()) _channel->StartWriteEventMonitor();
        });
    }
    void Send(std::vector<uint8_t>&& data) { Send(data); }
    void Send(std::string& str) { Send(std::vector<uint8_t>(str.begin(), str.end())); }
    void Send(std::string&& str) { Send(str); }
    
    // 关闭连接
    void CloseConnection()
    {
        _loop->RunInEvevtLoop([&]() -> void {
            _status = DISCONNECTING;
            if (_in_buffer.readable_size() > 0)
            {
                if (_message_handle)
                    _message_handle(shared_from_this(), _in_buffer);
            }
            if (_out_buffer.readable_size() > 0)
            {
                if (!_channel->isWriteable())
                    _channel->StartWriteEventMonitor();
            }
            if (_out_buffer.readable_size() == 0)
            {
                Release();
            }
        });
    }

    // 开启非活跃连接释放
    // 定时 timeout_sec 后释放连接
    void EnableInactiveRelease(Timer::time_t timeout_sec)
    {
        _loop->RunInEvevtLoop([this, timeout_sec]() -> void {
            _is_inactive_connection = true;

            // 存在，刷新
            if (_loop->HaveTimerTask(_connection_id))
                return _loop->FlashTimerTask(_connection_id);
            
            // 不存在，添加
            LOG(DEBUG, "timeout_sec:%d", timeout_sec);
            _loop->AddTimerTask(_connection_id, timeout_sec, [&]() -> void {
                Release();
            });
        });
        
    }
    
    // 关闭非活跃连接释放
    void CloseInactiveRelease()
    {
        _loop->RunInEvevtLoop([&]() -> void {
            _is_inactive_connection = false;
            if (_loop->HaveTimerTask(_connection_id))
                _loop->CancelTimerTask(_connection_id);
        });
    }

    // 切换协议
    void SwitchProtocols(const std::any& context, ConnectHandle connect_handle, MessageHandle message_handle, CloseHandle close_handle, EventHandle event_handle)
    {
        _loop->AssertInThread();
        _loop->RunInEvevtLoop([&]() -> void {
            SetContext(context);
            SetConnectHandle(connect_handle);
            SetMessageHandle(message_handle);
            SetCloseHandle(close_handle);
            SetEventHandle(event_handle);
        });
    }
};

class Acceptor
{
    using AcceptHandle = std::function<void(int)>;
private:
    Socket _socket;     // 用于创建监听套接字
    std::unique_ptr<Channel> _channel;   // 对监听套接字进行事件管理
    EventLoop* _loop;   // 对监听套接字进行事件监控
    AcceptHandle _call_back;

private:
    void ReadHandle()
    {
        int newfd = _socket.Accept();
        if (newfd < 0)
        {
            return;
        }

        if (_call_back) 
            _call_back(newfd);
    }

public:
    Acceptor(EventLoop* loop, uint16_t port)
        : _loop(loop)
    {
        _socket.CreateServer(port);
        _channel = std::unique_ptr<Channel>(new Channel(_socket.getFd(), _loop));
        _channel->SetReadHandle([this]() -> void {
            ReadHandle();
        });
        _channel->StartReadEventMonitor();
    }

    void SetAcceptCallBack(const AcceptHandle& call_back) { _call_back = call_back;  }
    void Listen() { _channel->StartReadEventMonitor(); }

};

class EventLoopThread
{
private:
    // max_delay_time = TimerWheel::DEFAULT_DELAY_TIME
    TimerTaskType::time_t _max_delay_time;
    std::mutex _mutex;
    std::condition_variable _cond;
    EventLoop* _loop;
    std::thread _thread;
private:

    void Start()
    {
        EventLoop loop(_max_delay_time);
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        _loop->StartLoop();
    }

public:
    EventLoopThread(TimerTaskType::time_t max_delay_time = TimerWheel::DEFAULT_DELAY_TIME)
        : _max_delay_time(max_delay_time)
        , _loop(nullptr)
        , _thread(std::thread([this](){ Start(); }))
    {}

    EventLoop* getEventLoop()
    {
        EventLoop* loop_ptr = nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&]() -> bool { return _loop != nullptr; });
            loop_ptr = _loop;
        }
        return loop_ptr;
    }

};

class EventLoopThreadPool
{
private:
    size_t _counds;
    size_t _index;
    TimerTaskType::time_t _max_delay_time;
    EventLoop* _base_loop;
    std::vector<std::unique_ptr<EventLoopThread>> _threads;
    std::vector<EventLoop*> _loops;
public:
    EventLoopThreadPool(EventLoop* base_loop, size_t n = 0, TimerTaskType::time_t max_delay_time = TimerWheel::DEFAULT_DELAY_TIME)
        : _counds(n)
        , _index(0)
        , _max_delay_time(max_delay_time)
        , _base_loop(base_loop)
    {
        if (_counds > 0)
        {
            _threads.resize(_counds);
            _loops.resize(_counds);
            for (int i = 0; i<_counds; ++i)
            {
                _threads[i] = std::unique_ptr<EventLoopThread>(new EventLoopThread(_max_delay_time));
                _loops[i] = _threads[i]->getEventLoop();
            }
        }
    }

    EventLoop* nextEventLoop()
    {
        if (_counds == 0)
            return _base_loop;
        _index = (_index + 1) % _counds;
        return _loops[_index];
    }
    
};

class TcpServer
{
    using ConnectHandle = Connection::ConnectHandle;
    using MessageHandle = Connection::MessageHandle;
    using CloseHandle = Connection::CloseHandle;
    using EventHandle = Connection::EventHandle;

private:
    uint64_t _next_id;  // 连接ID（自动增长）
    bool _enable_inactive;
    Timer::time_t _timeout_sec;


    std::unique_ptr<EventLoop> _base_loop;
    std::unique_ptr<Acceptor> _acceptor;
    EventLoopThreadPool _pool;
    std::unordered_map<uint64_t, Connection::ConnectionPtr> _connections;

    ConnectHandle _connected_handle;
    MessageHandle _message_handle;
    CloseHandle _close_handle;
    EventHandle _event_handle;
    CloseHandle _server_close_handle;
    // std::any _context;
private:
    void CreateConnetion(int fd)
    {
        ++_next_id;
        Connection::ConnectionPtr conn(new Connection(_pool.nextEventLoop(), _next_id, fd));
        conn->SetMessageHandle(_message_handle);
        conn->SetConnectHandle(_connected_handle);
        conn->SetCloseHandle(_close_handle);
        conn->SetEventHandle(_event_handle);
        conn->SetServerCloseHandle(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));

        if (_enable_inactive)
        {
            LOG(DEBUG, "_enable_inactive is true, _timeout_sec:%d", _timeout_sec);
            conn->EnableInactiveRelease(_timeout_sec);
        }
        conn->Established();

        _connections.insert({_next_id, conn});
    }
    void RemoveConnection(Connection::ConnectionPtr conn)
    {
        _base_loop->RunInEvevtLoop([&]() ->void {
            auto id = conn->getId();
            auto it = _connections.find(id);
            if (it != _connections.end())
            {
                _connections.erase(it);
            }
        });
    }

public:
    TcpServer(uint16_t port, size_t thread_size = 0, TimerTaskType::time_t max_delay_time = TimerWheel::DEFAULT_DELAY_TIME)
        : _next_id(0)
        , _enable_inactive(false)
        , _base_loop(std::unique_ptr<EventLoop>(new EventLoop))
        , _acceptor(std::unique_ptr<Acceptor>(new Acceptor(_base_loop.get(), port)))
        , _pool(_base_loop.get(), thread_size, max_delay_time)
    {
        _acceptor->SetAcceptCallBack(std::bind(&TcpServer::CreateConnetion, this, std::placeholders::_1));
        signal(SIGPIPE, SIG_IGN); // 防止连接断开还有数据在发送，进而导致进程终止
    }

    void SetConnectHandle(ConnectHandle connected_handle) { _connected_handle = connected_handle; }
    void SetMessageHandle(MessageHandle message_handle) { _message_handle = message_handle; }
    void SetCloseHandle(CloseHandle close_handle) { _close_handle = close_handle; }
    void SetEventHandle(EventHandle event_handle) { _event_handle = event_handle; }
    // void SetContext(std::any context) { _context.reset(); _context = context; }

    // 启动非活跃连接销毁
    void EnableInactiveRelease(Timer::time_t timeout_sec) {
        _timeout_sec = timeout_sec;
        _enable_inactive = true;
    }

    // 添加定时任务
    void AddScheduledTask(const Timer::TaskFunc& func, Timer::time_t timeout_sec)
    {
        _base_loop->RunInEvevtLoop([&]() -> void {
            ++_next_id;
            _base_loop->AddTimerTask(_next_id, timeout_sec, func);
        });
    }

    void Start()
    {
        _acceptor->Listen();
        _base_loop->StartLoop();
    }
};

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
/*
    Socket
*/
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

// 创建套接字
bool Socket::Create()
{
    int sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (sockfd == ERROR_SOCKET)
    {
        // THROW_EXCEPTION(ERROR, "socket 创建失败");
        LOG(ERROR, "socket 创建失败");
        return false;
    }
    _sockfd = sockfd;
    return true;
}

bool Socket::Bind(const std::string& ip, uint16_t port)
{
    struct sockaddr_in addr;
    socklen_t len = sizeof(struct sockaddr_in);

    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = inet_addr(ip.c_str());

    int ret = bind(_sockfd, (sockaddr*)&addr, len);
    if (ret < 0)
    {
        LOG(ERROR, "bind 失败");
        return false;
    }
    return true;
}

// 监听套接字
bool Socket::Listen(uint32_t backlog)
{
    // int listen(int sockfd, int backlog);
    int ret = listen(_sockfd, backlog);
    if (ret < 0)
    {
        LOG(ERROR, "listen 失败");
        return false;
    }
    return true;
}

// 向服务端发起连接
bool Socket::Connect(const std::string& ip, uint16_t port)
{
    struct sockaddr_in addr;
    socklen_t len = sizeof(struct sockaddr_in);

    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = inet_addr(ip.c_str());

    int ret = connect(_sockfd, (sockaddr*)&addr, len);
    if (ret < 0)
    {
        LOG(ERROR, "connect 失败");
        return false;
    }
    return true;
}

// 获取新连接
int Socket::Accept()
{
    int new_fd = accept(_sockfd, nullptr, nullptr);
    if (new_fd == ERROR_SOCKET)
    {
        LOG(ERROR, "Accpet 创建失败: ", strerror(errno));
        // throw Exception("Socket Accept error");
    }

    return new_fd;
}

// 接收数据
ssize_t Socket::Recv(void *buf, size_t len, int flags)
{
    ssize_t ret = recv(_sockfd, buf, len, flags);
    if (ret <= 0)
    {
        if (errno == EAGAIN || errno == EINTR)
            return 0;

        LOG(ERROR, "recv 失败");
        return -1;
    }
    LOG(DEBUG, "recv 接收完成");

    return ret;
}

// 发送数据
ssize_t Socket::Send(const void *buf, size_t len, int flags)
{
    ssize_t ret = send(_sockfd, buf, len, flags);
    if (ret < 0)
    {
        if (errno == EAGAIN || errno == EINTR)
            return 0;
        LOG(ERROR, "send 失败");
        return -1;
    }
    LOG(DEBUG, "send 发送完成");
    return ret;
}

bool Socket::CreateServer(uint16_t port, const std::string& ip, bool setNoBlock)
{
    if (!Create()) return false;
    ReuseAddress(); // 设置端口重用
    if (setNoBlock) NonBlock();
    if (!Bind(ip, port)) return false;
    if (!Listen()) return false;
    return true;
}

bool Socket::CreateClient(uint16_t port, const std::string& ip)
{
    if (!Create()) return false;
    if (!Connect(ip, port)) return false;

    return true;
}

void Socket::ReuseAddress()
{
    int optval = 1; 
    setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (const void*)&optval, sizeof(int));
    optval = 1; 
    setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (const void*)&optval, sizeof(int));
}

void Socket::NonBlock()
{
    int flag = fcntl(_sockfd, F_GETFL);
    fcntl(_sockfd, F_SETFD, flag | O_NONBLOCK);
}


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
/*
    Channel 
*/
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

// 移除事件监控
void Channel::RemoveEvent() { _loop->RemoveEvent(this); }

// 更新事件
void Channel::UpdateEvent() {  _loop->UpdateEvent(this); }

// 触发事件调用
void Channel::TriggerEvent()
{
    if (_any_handle) _any_handle();
    if (
        (_trigger_events & EPOLLIN)
        || (_trigger_events & EPOLLRDHUP)   // 流套接字对等方关闭连接，或关闭写入一半连接。
        || (_trigger_events & EPOLLPRI)     // 文件描述符上存在例外情况。
    )
    {
        // if (_any_handle) _any_handle(); // any_handle 后执行保持活跃度
        if (_read_handle) _read_handle();
    }
    else if (_trigger_events & EPOLLOUT)
    {
        //if (_any_handle) _any_handle();
        if (_write_handle) _write_handle();
        
    }
    else if (_trigger_events & EPOLLERR)
    {
        
        //if (_any_handle) _any_handle(); // any_handle 先执行，确保错误前执行完毕
        if (_error_handle) _error_handle();
    }
    else if (_trigger_events & EPOLLHUP)
    {
        //if (_any_handle) _any_handle();
        if (_close_handle) _close_handle();
    }
}

// 设置事件监控

// 启动读事件监控
void Channel::StartReadEventMonitor() { _monitor_events |= EPOLLIN; UpdateEvent(); }
// 启动写事件监控
void Channel::StartWriteEventMonitor() { _monitor_events |= EPOLLOUT; UpdateEvent(); }
// 关闭读事件监控
void Channel::CloseReadEventMonitor() { _monitor_events &= ~EPOLLIN; UpdateEvent(); }
// 关闭写事件监控
void Channel::CloseWriteEventMonitor() { _monitor_events &= ~EPOLLOUT; UpdateEvent(); }
// 关闭所有事件监控
void Channel::CloseEventMonitor() { _monitor_events = 0; RemoveEvent(); }

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
/*
    Poller 
*/
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

void Poller::Poll(std::vector<Channel*>& active)
{
    // LOG(DEBUG, "Poll in");
    int ready_fds = epoll_wait(_epoll_fd, _events.data(), MAX_EPOLL_EVENTS, -1);
    if (ready_fds < 0)
    {
        if (errno == EINTR) 
            return;
        LOG(FATAL, "epoll_wait error:", strerror(errno));
    }
    for (int i = 0; i < ready_fds; ++i)
    {
        auto it = _channels.find(_events[i].data.fd);
        if (it == _channels.end())
            LOG(FATAL, "channels 未找到对应信息");
        it->second->SetTriggerEvents(_events[i].events);
        active.push_back(it->second);
    }
}

void Poller::RemoveEvent(Channel* channel) noexcept
{
    if (!haveChannel(channel)) return;
    _channels.erase(channel->getFd());
    PollerCtl(channel, EPOLL_CTL_DEL);
}

void Poller::UpdateEvent(Channel* channel) noexcept
{
    if (!haveChannel(channel))
    {
        // _channels[channel->getFd()] = channel;
        _channels.insert({channel->getFd(), channel});
        PollerCtl(channel, EPOLL_CTL_ADD);
        return;
    }
    PollerCtl(channel, EPOLL_CTL_MOD);
}

void Poller::PollerCtl(Channel* channel, int op) noexcept
{
    int fd = channel->getFd();
    struct epoll_event event;
    event.data.fd = fd;
    event.events = channel->getMonitorEvents();
    epoll_ctl(_epoll_fd, op, fd, &event);
}

bool Poller::haveChannel(Channel* channel) noexcept
{
    auto it = _channels.find(channel->getFd());
    if (it == _channels.end())
    {
        return false;
    }
    return true;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
/*
    EventLoop 
*/
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

EventLoop::EventLoop(TimerTaskType::time_t max_delay_time)
    : _event_fd(CreateEventFd())
    , _thread_id(std::this_thread::get_id())
    , _event_channel(new Channel(_event_fd, this))
    , _timer(this, max_delay_time)
{
    LOG(DEBUG, "EventLoop init");
    _event_channel->SetReadHandle(std::bind(&EventLoop::ReadeEventFd, this));
    _event_channel->StartReadEventMonitor();
}

int EventLoop::CreateEventFd()
{
    int efd = eventfd(0, EFD_CLOEXEC);
    if (efd < 0)
    {
        LOG(ERROR, "eventfd failed");
        abort();
    }

    return efd;
}

void EventLoop::ReadeEventFd()
{
    uint64_t res = 0;
    ssize_t ret = read(_event_fd, &res, sizeof(res));
    if (ret < 0)
    {
        if (errno == EINTR || errno == EAGAIN)
            return;
        
        LOG(ERROR, "read event error");
        abort();
    }
}

void EventLoop::WeakUpEventFd()
{
    uint64_t res = 0;
    ssize_t ret = write(_event_fd, &res, sizeof(res));
    if (ret < 0)
    {
        if (errno == EINTR)
            return;
        
        LOG(ERROR, "write event error");
        abort();
    }
}

void EventLoop::RunAllTask()
{
    std::vector<func_t> funcs;
    {
        std::unique_lock<std::mutex> lock(_mutex);
        funcs.swap(_task_queue);
    }
    for (auto& f : funcs)
    {
        f();
    }
}

void EventLoop::StartLoop()
{
    while (true)
    {
        // 事件监控
        std::vector<Channel*> actives;
        _poller.Poll(actives);

        // 事件处理
        for (auto& channel : actives)
        {
            channel->TriggerEvent();
        }

        // 执行任务
        RunAllTask();
    }
}

void EventLoop::RunInEvevtLoop(const func_t& handle)
{
    if (IsEventLoopThread())
    {
        return handle();
    }
    else 
    {
        return PushIntoTaskQueue(handle);
    }
}

// 将任务压入任务队列
void EventLoop::PushIntoTaskQueue(const func_t& task)
{
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _task_queue.push_back(task);
    }
    WeakUpEventFd();
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
/*
    TimerWheel 
*/
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

void TimerWheel::add(taskid_t id, time_t delay_time, const TaskFunc& call_back)
{
    // 如果 delay_time > _maxest_delay_time 抛异常
    if (delay_time > _maxest_delay_time)
    {
        LOG(ERROR, "延迟时间过长, delay_time:%d maxest_delay_time:%d", delay_time, _maxest_delay_time);
        abort();
    }
    TaskPtr task_ptr(new TimerTask(id, delay_time, call_back));
    task_ptr->setRelease(std::bind(&TimerWheel::RemoveTimer, this, id));
    _timers[id] = WeakTaskPtr(task_ptr);
    _time_wheel[(delay_time + _tick)%_maxest_delay_time].push_back(task_ptr);
}


void TimerWheel::flash(taskid_t id)
{
    auto it = _timers.find(id);
    if (it == _timers.end())
    {
        return;
    }
    TaskPtr pt = it->second.lock();
    time_t delay_time = pt->getDelayTime();
    _time_wheel[(delay_time + _tick)%_maxest_delay_time].push_back(pt);
}

void TimerWheel::RemoveTimer(taskid_t id)
{
    _timers.erase(id);
}

void TimerWheel::runTick()
{
    _tick = (_tick + 1) % _maxest_delay_time;
    _time_wheel[_tick].clear();
    //std::cout << _tick << std::endl;
}

void TimerWheel::canceledTimer(taskid_t id)
{
    auto it = _timers.find(id);
    if (it == _timers.end())
    {
        return;
    }

    auto task_ptr = it->second.lock();
    if (task_ptr != nullptr) 
    {
        task_ptr->canceled();
    }
}

bool TimerWheel::haveTimer(taskid_t id)
{
    auto it = _timers.find(id);
    if (it == _timers.end())
    {
        return false;
    }
    else
    {
        return true;
    }
}

int Timer::CraeteTimerFd()
{
    int fd = timerfd_create(CLOCK_MONOTONIC, 0);
    if (fd < -1)
    {
        LOG(FATAL, "timer_create error");
        abort();
    }

    struct itimerspec itime;
    itime.it_interval.tv_sec = 1;
    itime.it_interval.tv_nsec = 0;
    itime.it_value.tv_sec = 1;
    itime.it_value.tv_nsec = 0;

    timerfd_settime(fd, 0, &itime, nullptr);

    return fd;
}

void Timer::ReadTimerFd()
{
    uint64_t times = 0;
    ssize_t ret = read(_timerfd, &times, sizeof(times));
    if (ret < 0) 
    {
        LOG(FATAL, "timer read error");
        close(_timerfd);
        abort();
    }
}

// 增加定时任务
void Timer::AddTimerTask(taskid_t id, time_t delay_time, const TaskFunc& handle) {
    _loop->RunInEvevtLoop([&]() -> void {
        _timer_wheel.add(id, delay_time, handle);
    });
}

// 刷新定时任务
void Timer::FlashTimerTask(taskid_t id) {
    _loop->RunInEvevtLoop([&]() -> void {
        _timer_wheel.flash(id);
    });
}

// 取消定时任务
void Timer::CancelTimerTask(taskid_t id) { 
    _loop->RunInEvevtLoop([&]() -> void {
        _timer_wheel.canceledTimer(id);
    });
}

bool Timer::HaveTimerTask(taskid_t id)
{
    return _timer_wheel.haveTimer(id);
}